package DianShang_2024.ds_01.extract

import org.apache.spark.sql.SparkSession

import java.util.Properties

object extract01 {
  def main(args: Array[String]): Unit = {
    /*
          抽取shtd_store库中user_info的增量数据进入Hive的ods库中表user_info。根据ods.user_info表中operate_time或create_time作为增量字段
          (即MySQL中每条数据取这两个时间中较大的那个时间作为增量字段去和ods里的这两个字段中较大的时间进行比较)，只将新增的数据抽入，字段名称、类型不变，
          同时添加静态分区，分区字段为etl_date，类型为String，且值为当前比赛日的前一天日期（分区字段格式为yyyyMMdd）。使用hive cli执行show partitions ods.user_info命令
     */
    //  准备sparksql的环境
    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("抽取数据第一题")
      .enableHiveSupport()
      .getOrCreate()

    spark.sql("use ods")

    //  配置连接mysql的配置
    val jdbc_conf=new Properties()
    jdbc_conf.setProperty("user","root")
    jdbc_conf.setProperty("password","123456")
    jdbc_conf.setProperty("driver","com.mysql.jdbc.Driver")

    //  连接mysql拿到数据源的数据，并且创建临时表
    spark.read.jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false","user_info",jdbc_conf)
      .createOrReplaceTempView("user_info_temp")

    //  拿到hive数据里面最大的时间去和mysql里面的数据进行比较
    /*
              1.collect(): collect() 是一个 Action 操作，它会触发 Spark 作业的执行，并将 DataFrame 或 Dataset 的所有数据以数组的形式返回给
              驱动程序。在这里，collect() 将查询结果收集到一个数组中。
              2.(0): 一旦调用 collect() 方法后，我们得到了一个包含查询结果的数组。(0) 表示取这个数组的第一个元素（索引为 0）
              3.get(0): 一旦我们获取到了数组中的第一个元素，再调用 get(0) 可以获取这个元素中的第一个值（索引为 0）。

     */
    val max_time=spark.sql(
      """
        |select
        |if(operate_time > create_time,operate_time,create_time) as time
        |from ods.user_info2
        |order by time desc
        |limit 1
        |""".stripMargin).collect()(0).get(0).toString

    //  根据条件抽取数据
    /*
          根据逆向排序的方法将最大的时间排在第一行然后coolect得到dataframe进行操作，逐级拿到
     */
    spark.sql(
      s"""
        |insert into table ods.user_info2
        |partition(etl_date='20231017')
        |select
        |*
        |from user_info_temp
        |where if(operate_time>create_time,operate_time,create_time) > cast('$max_time' as timestamp)
        |""".stripMargin)





      //  关闭环境
    spark.close()
  }

}
